package com.lin.fengxiu.manager.rocketmq;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @ClassName ProducerSchedule
 * @Description TODO 生产者
 * @Author wushaopei
 * @Date 2021/4/16 6:38
 * @Version 1.0
 */
@Component
public class ProducerSchedule {

    // 生产者类
    private DefaultMQProducer producer;

    @Value("${rocketmq.producer.producer-group}")
    private String producerGroup;

    @Value("${rocketmq.namesrv-addr}")
    private String namesrvAddr;

    public ProducerSchedule(){

    }

    /**
     * @Description TODO 实例化生产者
     */
    @PostConstruct
    public void defaultMQProducer(){

        if(this.producer == null){
            // 实例化消息生产者Producer
            this.producer = new DefaultMQProducer(this.producerGroup);
            // 这个是为Producer设置NameServer的地址，让他可以拉取路由信息
            // 这样才知道每个Topic的数据分散在哪些Broker机器上
            // 然后才可以把消息发送到Broker上去
            this.producer.setNamesrvAddr(this.namesrvAddr);
        }
        try {
            // 这里启动一个Producer
            this.producer.start();
            System.out.println("--------producer start--------");
        }catch (MQClientException e){
            e.printStackTrace();
        }
    }

    /**
     * @Description TODO 主方法：生产者发送消息用的
     * @param topic
     * @param messageText
     * @return
     */
    public String send(String topic,  String messageText) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        // 这里是构建一条消息对象
        Message message = new Message(topic,messageText.getBytes());
        // 延迟消息的级别： messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m
        message.setDelayTimeLevel(4); // 4 的级别是延迟 30s

        // 发送消息并接收返回结果，可以不接收
        SendResult result = this.producer.send(message);
        System.out.println(result.getMsgId());
        System.out.println(result.getSendStatus());
        // 返回消息的Id
        return result.getMsgId();
    }

}
